package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.commont.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/4/25 13:48
 */
public class Dwd_05_DwdTradeOrderCancelDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_05_DwdTradeOrderCancelDetail().init(
            30005,
            2,
            "Dwd_05_DwdTradeOrderCancelDetail"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        
        // 1. 读取 dwd 层的下单事务事实表
        tEnv.executeSql(
            "create table dwd_trade_order_detail(  " +
                "id string,  " +
                "order_id string,  " +
                "user_id string,  " +
                "sku_id string,  " +
                "sku_name string,  " +
                "province_id string,  " +
                "activity_id string,  " +
                "activity_rule_id string,  " +
                "coupon_id string,  " +
                "date_id string,  " +
                "create_time string,  " +
                "sku_num string,  " +
                "split_original_amount string,  " +
                "split_activity_amount string,  " +
                "split_coupon_amount string,  " +
                "split_total_amount string,  " +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "dwd_trade_order_detail_1"));
        //tEnv.sqlQuery("select * from dwd_trade_order_detail").execute().print();
        // 2. 读取 ods_db 数据
        readOdsDb(tEnv, "dwd_trade_order_detail_2");
        // 3. 从 ods_db 中过滤出 order_info 中的订单取消数据
        Table orderInfoCancel = tEnv.sqlQuery("select " +
                                        " `data`['id'] id," +
                                        " `data`['operate_time'] order_cancel_time, " +
                                        " ts " +
                                        "from ods_db " +
                                        "where `database`='gmall2023' " +
                                        "and `table`='order_info' " +
                                        "and `type`='update' " +
                                        "and `old`['order_status']='1001' " +
                                        "and `data`['order_status']='1003' ");
        
        tEnv.createTemporaryView("order_info_cancel", orderInfoCancel);
        
        // 4. 两张表的 join
        Table result = tEnv.sqlQuery("select " +
                                         "od.id,  " +
                                         "od.order_id,  " +
                                         "od.user_id,  " +
                                         "od.sku_id,  " +
                                         "od.sku_name,  " +
                                         "od.province_id,  " +
                                         "od.activity_id,  " +
                                         "od.activity_rule_id,  " +
                                         "od.coupon_id,  " +
                                         "date_format(oi.order_cancel_time, 'yyyy-MM-dd') order_cancel_date_id,  " +
                                         "oi.order_cancel_time,  " +
                                         "od.sku_num,  " +
                                         "od.split_original_amount,  " +
                                         "od.split_activity_amount,  " +
                                         "od.split_coupon_amount,  " +
                                         "od.split_total_amount,  " +
                                         "oi.ts   " +
                                        "from order_info_cancel oi " +
                                        "join dwd_trade_order_detail od " +
                                        "on oi.id=od.order_id ");
        // 5. 写出到 dwd 层
        tEnv.executeSql("create table dwd_trade_cancel_detail(  " +
                                "id string,  " +
                                "order_id string,  " +
                                "user_id string,  " +
                                "sku_id string,  " +
                                "sku_name string,  " +
                                "province_id string,  " +
                                "activity_id string,  " +
                                "activity_rule_id string,  " +
                                "coupon_id string,  " +
                                "date_id string,  " +
                                "cancel_time string,  " +
                                "sku_num string,  " +
                                "split_original_amount string,  " +
                                "split_activity_amount string,  " +
                                "split_coupon_amount string,  " +
                                "split_total_amount string,  " +
                                "ts bigint  " +
                                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CANCEL_DETAIL));
        result.executeInsert("dwd_trade_cancel_detail");
        
    }
}
/*
订单取消事务事实表
    
    也需要和下单事务事实表同样的数据:
        order_detail:    insert
        
        order_info:      update    order_status: 1001->1003
        
        order_detail_activity  insert
        
        order_detail_coupon    insert
        
        
    ttl:
        30min + 5s


    2张表:
        1. 下单事务事实表: 把详情和详情活动和详情优惠券已经 join 号
        
        2. 从 order_info 过滤出取消的订单
        
    dwd_trade_order_detail
        join
    order_info_cancel
    
    
*/